use std::sync::Arc;

use crate::error::ServerError;
use crate::protocol::data_schema::DataTrait;
use crate::protocol::type_schema::TypeSchema;
use crate::abstractions::data::Data;
use crate::services::manager::SERVICE_MANAGER;
use crate::tasks::task::Task;
use crate::workflows::portable_run::get_available_port;
use crate::workflows::external_task_function::workflow_start_task;

pub type TaskFunc = Arc<dyn Fn(Data, u16) -> Result<Data, ServerError> + Send + Sync>;

/// 格式：（服务ID，任务名）
pub type TaskId = (String, String);

pub enum TaskPortableAbs {
    Simple(String, TypeSchema, TypeSchema, TaskFunc, u64, Vec<(u64, String)>),
    Builtin(String, TypeSchema, TypeSchema, TaskFunc, u64, Vec<(u64, String)>)
}

use TaskPortableAbs::*;

use super::task::TaskAbstraction;

pub fn zmqsend(serial_id: u64, port: u16) {
    if port == 0 {
        return;
    }
    let context = zmq::Context::new();
    let socket = context.socket(zmq::PUSH).unwrap();
    let endpoint = format!("tcp://127.0.0.1:{}", port);
    socket.connect(&endpoint).unwrap();
    socket.send(format!("{}", serial_id).as_bytes(), 0).unwrap();
}

pub fn get_new_serial_id() -> u64 {
    static SERIAL_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(100);
    SERIAL_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
}

impl TaskPortableAbs {
    pub fn from_taskabs(task: Arc<TaskAbstraction>) -> Result<TaskPortableAbs, ServerError> {
        if task.is_builtin() {
            let (name, tin, tout, fun) = task.flatten_clone();
            let serial_id = get_new_serial_id();
            let task_func = Arc::new(move|input: Data, port: u16| {
                fun(input)
            });
            Ok(TaskPortableAbs::Builtin(name, tin, tout, task_func, serial_id, vec![]))
        }
        else {
            let service_id = Arc::new(task.service_id().clone());
            let handle = std::thread::spawn(move || {
                let rt = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .unwrap();
                let service = rt.block_on(async {
                    let mut manager = SERVICE_MANAGER.get().unwrap().lock().await;
                    let mut service = manager.get_service(&service_id.clone()).unwrap();
                    service.mount().await?;
                    manager.renew_service(&service);
                    drop(manager);
                    Ok(Arc::new(service))
                });
                service
            });
            let service = handle.join().unwrap()?;
            let task_name_clone = Arc::new(task.task_name().clone());
            let serial_id = get_new_serial_id();
            let task_func = Arc::new(move|input: Data, port: u16| {
                workflow_start_task(port, task_name_clone.clone(), input, service.clone())
            });
            Ok(TaskPortableAbs::Simple(task.name().clone(), task.tin().clone(), task.tout().clone(), task_func, serial_id, vec![]))
        }
    }
}

impl TaskPortableAbs {
    pub fn is_builtin(&self) -> bool {
        match self {
            Simple(_, _, _, _, _, _) => false,
            Builtin(_, _, _, _, _, _) => true
        }
    }

    pub fn name(&self) -> &String {
        match self {
            Simple(name, _, _, _, _, _) => name,
            Builtin(name, _, _, _, _, _) => name
        }
    }

    pub fn tin(&self) -> &TypeSchema {
        match self {
            Simple(_, tin, _, _, _, _) => tin,
            Builtin(_, tin, _, _, _, _) => tin
        }
    }

    pub fn tout(&self) -> &TypeSchema {
        match self {
            Simple(_, _, tout, _, _, _) => tout,
            Builtin(_, _, tout, _, _, _) => tout
        }
    }

    pub fn fun(&self) -> TaskFunc {
        match self {
            Simple(_, _, _, fun, _, _) => fun.clone(),
            Builtin(_, _, _, fun, _, _) => fun.clone()
        }
    }

    pub fn sendfun(&self) -> TaskFunc {
        let fun = self.fun().clone();
        let serial_id = self.serial_id();
        Arc::new(move |input: Data, port: u16| {
            let result = fun.clone()(input, port);
            zmqsend(serial_id, port);
            result
        })
    }

    pub fn serial_id(&self) -> u64 {
        match self {
            Simple(_, _, _, _, serial_id, _) => *serial_id,
            Builtin(_, _, _, _, serial_id, _) => *serial_id
        }
    }

    pub fn signals(&self) -> &Vec<(u64, String)> {
        match self {
            Simple(_, _, _, _, _, signals) => signals,
            Builtin(_, _, _, _, _, signals) => signals
        }
    }

    pub fn flatten(&self) -> (String, TypeSchema, TypeSchema, TaskFunc, u64, Vec<(u64, String)>) {
        match self {
            Simple(name, tin, tout, fun, serial_id, signals) => (name.clone(), tin.clone(), tout.clone(), fun.clone(), *serial_id, signals.clone()),
            Builtin(name, tin, tout, fun, serial_id, signals) => (name.clone(), tin.clone(), tout.clone(), fun.clone(), *serial_id, signals.clone())
        }
    }   
}